package com.gao.mq.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gao.mq.constant.Constants;
import com.gao.mq.domain.MessageInfo;
import com.gao.mq.dto.ContentModel;
import com.gao.mq.handler.IMessageHandler;
import com.gao.mq.handler.factory.MessageHandlerConfig;
import com.gao.mq.handler.factory.MessageHandlerFactory;
import com.gao.mq.util.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Future;

/**
 * 消息消费者
 */
@Service
public class MessageConsumerService implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(MessageConsumerService.class);

    @Autowired
    private RedisUtil redisUtil;

    @Autowired
    private MessageHandlerFactory messageHandlerFactory;

    @Autowired
    private ApplicationContext context;

    @Autowired
    private ThreadPoolTaskExecutor threadTaskExecutor;

    @Autowired
    private MessageHandlerFactory handlerConfig;

    /**
     * 在所有的 Bean 初始化之后调用
     */
    @Override
    public void afterPropertiesSet() {
        logger.info("============== Redis 消息队列 Start ================");
        Future<?> future = threadTaskExecutor.submit(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    // 遍历多个消息类型
                    Set<Integer> msgCodes = handlerConfig.msgCodeSet();
                    for (Integer code : msgCodes) {
                        // 获取队列中的信息
                        //String msg = (String) redisUtil.rightPop(Constants.RedisMsgKey.REDIS_MSG_KEY(code), 1);
                        MessageInfo messageInfo = (MessageInfo)redisUtil.rightPop(Constants.RedisMsgKey.REDIS_MSG_KEY(code), 1);
                        //JSONObject messageInfo = JSONObject.parseObject(msg);
                        //MessageInfo messageInfo = JSONObject.parseObject(msg, MessageInfo.class);
                        // 发送消息给用户
                        if (messageInfo != null) {
                            // 从工厂中获取消息处理器
                            logger.info("处理{}消息中，{}", Constants.MessageType.getEnumByCode(code).getDesc() ,JSON.toJSONString(messageInfo));
                            IMessageHandler handler = messageHandlerFactory.getMessageHandler(code);
                            handler.doHandle(messageInfo);
                        }
                    }
                }
            }
        });
        //返回结果
        JSONObject jsonResult = new JSONObject();
        try{
            //线程调度结束时，才获取结果
            if (future.isDone()){
                jsonResult = (JSONObject) future.get();
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
